package com.zengqingfa.kafka.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @author zengqingfa
 * @className TimeInterceptor
 * @description 给消息增加一个时间戳
 * @create 2021/8/2 11:30
 */
public class TimeInterceptor implements ProducerInterceptor<String, String> {
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        //取出数据
        String value = record.value();
        //创建一个新的ProducerRecord对象 并返回
        ProducerRecord<String,String> rec=new ProducerRecord<>(record.topic(),record.partition(), record.key(), System.currentTimeMillis()+","+value);
        return rec;
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}